#! /usr/bin/env python
#
# MapReduce.py
#

from code import step1
from code import step2
from step2 import Struct
from code.giss_data import BASE_YEAR
from code import parameters

from tool import giss_io

# http://packages.python.org/mrjob/
from mrjob.job import MRJob


class GissAnalysisMapReduce(MRJob):
    
    DEFAULT_PROTOCOL = 'pickle'
    
    def __init__(self, *args, **kwargs):
        super(GissAnalysisMapReduce, self).__init__(*args, **kwargs)
        
  
    def get_data(self, _, line):
        
        """Extract the 12-digit station record identifier."""        
        yield line[:12], line
  
  
    def combine_data(self, id, lines):
 
        """ Records are grouped into blocks of lines, 
            all of which share the same 12-digit ID.
            Yield the 11-Digit station Id and the Series 
            object corresponding to that station ID.
        """
        for record in giss_io.V2MapReduceMeanReader(id, sorted(lines, key=lambda r: r[:16])):
            yield record.uid[:11], record

  
    def reduce_data(self, key, value):
     
        records = step1.comb_records(value)
        helena_adjusted = step1.adjust_helena(records)
        combined_pieces = step1.comb_pieces(helena_adjusted)
        without_strange = step1.drop_strange(combined_pieces)
        records = list(step1.alter_discont(without_strange))
        i = 0
        for record in records:
            assert record.first_year == BASE_YEAR
            
            """Drop records which don't have at least one month index
            with at least *parameters.station_drop_minimum_months* valid data.
            """
            mmax = max(record.get_monthly_valid_counts())
            if mmax < parameters.station_drop_minimum_months:
               records.pop(i)
            i = i + 1
        
        all = step2.annotate_records(records)    

        for record in all:
            yield record.uid, record
            

    def adjust_urban_stations(self, key, records):
        
        adjusted = step2.urban_adjustments(records)
        for record in adjusted:
            yield record.uid, record

           
    def steps(self):
        return [
            self.mr(
                mapper=self.get_data,
                reducer=self.combine_data, 
            ),
            self.mr(
                reducer=self.reduce_data,
            ),
            self.mr(
                reducer=self.adjust_urban_stations
            )
        ]   

        
 
if __name__ == '__main__':
     GissAnalysisMapReduce.run()